package com.lsx143.wordcount.day2;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

import java.util.Arrays;

public class ConnectTest {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Integer> dss1 = env.fromCollection(Arrays.asList(1, 3, 5, 7, 9));
        DataStreamSource<Integer> dss2 = env.fromCollection(Arrays.asList(2, 4, 6, 8, 10));
        DataStreamSource<String> dss3 = env.fromCollection(Arrays.asList("jon", "all", "bit"));
        try {
//            dss1.connect(dss2)
//                    .map(new CoMapFunction<Integer, Integer, Integer>() {
//                        @Override
//                        public Integer map1(Integer value) throws Exception {
//                            return value;
//                        }
//
//                        @Override
//                        public Integer map2(Integer value) throws Exception {
//                            return value;
//                        }
//                    })
//                    .print();
            dss1.connect(dss3)
                    .map(new CoMapFunction<Integer, String, String>() {
                        @Override
                        public String map1(Integer value) throws Exception {
                            return value.toString();
                        }

                        @Override
                        public String map2(String value) throws Exception {
                            return value;
                        }
                    }).print();
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
